package com.stormdemo.topn.bolt;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

import backtype.storm.task.TopologyContext;
import backtype.storm.topology.BasicOutputCollector;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.topology.base.BaseBasicBolt;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Tuple;

/**
 * @Author: blogchong
 * @Blog: www.blogchong.com
 * @Mailbox: blogchong@163.com
 * @QQGroup: 191321336
 * @Weixin: blogchong
 * @Data: 2015/4/7
 */

@SuppressWarnings("serial")
public class PrintBolt extends BaseBasicBolt {
	private static Connection con = null;

	@Override
	public void prepare(Map stormConf, TopologyContext context) {
		String url = "jdbc:mysql://localhost:3306/test?user=root&password=root";
		try {
			con = DriverManager.getConnection(url);
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}

	public void execute(Tuple input, BasicOutputCollector collector) {
		try {
			String mesg = input.getString(0);
			if (mesg != null)
				System.out.println(mesg);
			saveTomysql(mesg, 1);
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		declarer.declare(new Fields("print"));
	}

	public static void saveTomysql(String msg, Integer age) {
		try {
			PreparedStatement stmt = con
					.prepareStatement("insert into test(name,age)values(?,?)");
			stmt.setString(1, msg);
			stmt.setInt(2, age);
			stmt.executeUpdate();
		} catch (SQLException e) {
			e.printStackTrace();
		}
	}
}
